package org.xydpeng.framework.l2cache.support;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
import org.springframework.boot.autoconfigure.cache.CacheProperties;
import org.springframework.cache.support.AbstractValueAdaptingCache;
import org.springframework.cache.support.NullValue;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.xydpeng.framework.l2cache.enums.CacheOperation;
import org.xydpeng.framework.l2cache.enums.RedisDataType;
import org.xydpeng.framework.l2cache.message.CacheMessage;
import org.xydpeng.framework.l2cache.message.CacheMessageMulticaster;
import org.xydpeng.framework.l2cache.properties.L2CacheConfigProperties;

import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;

/**
 * @author xydpeng
 * @version v1.0
 * @date: 2024/6/4
 * @description:
 */
public class RedisCaffeineL2Cache extends AbstractValueAdaptingCache {
    /**
     * redis緩存未命中，查询数据库时加锁
     */
    private final Lock lock = new ReentrantLock();

    private final String name;

    private final com.github.benmanes.caffeine.cache.Cache<Object, Object> cache;

    private final RedisTemplate redisTemplate;

    private CacheMessageMulticaster cacheMessageMulticaster;

    private L2CacheConfigProperties l2CacheConfigProperties;

    private L2CacheConfigProperties.L2CacheProperties l2CacheProperties;

    private CacheProperties cacheProperties;

    private RedisDataType redisDataType;

    private Duration redisTimeToLive;
    /**
     * Key prefix.
     */
    private String redisKeyPrefix;

    /**
     * Whether to use the key prefix when writing to Redis.
     */
    private boolean redisUseKeyPrefix = true;


    /**
     * Create an {@code AbstractValueAdaptingCache} with the given setting.
     */
    protected RedisCaffeineL2Cache(String name,
                                   CacheProperties cacheProperties,
                                   L2CacheConfigProperties l2CacheConfigProperties,
                                   RedisTemplate redisTemplate,
                                   CacheMessageMulticaster cacheMessageMulticaster) {
        super(l2CacheConfigProperties.getConfig(name).getCacheNullValues());
        this.name = name;
        this.cacheProperties = cacheProperties;
        this.l2CacheConfigProperties = l2CacheConfigProperties;
        this.l2CacheProperties = l2CacheConfigProperties.getConfig(name);
        this.redisTemplate = redisTemplate;
        this.cacheMessageMulticaster = cacheMessageMulticaster;
        this.redisDataType = this.l2CacheProperties.getRedisDataType();
        this.redisTimeToLive = this.l2CacheProperties.getTimeToLive();
        this.redisKeyPrefix = StringUtils.hasText(this.l2CacheProperties.getKeyPrefix()) ? this.l2CacheProperties.getKeyPrefix() : cacheProperties.getRedis().getKeyPrefix();
        this.redisUseKeyPrefix = Objects.isNull(this.l2CacheProperties.isUseKeyPrefix()) ? this.l2CacheProperties.isUseKeyPrefix() : cacheProperties.getRedis().isUseKeyPrefix();
        this.cache = createCaffeineCache();
    }

    private Cache<Object, Object> createCaffeineCache() {
        Long maximumSize = l2CacheProperties.getMaximumSize();
        Assert.isTrue(maximumSize > 0, "must config L1cache MaximumSize");
        Cache<Object, Object> cache = Caffeine.newBuilder()
                .maximumSize(maximumSize)
                .initialCapacity(l2CacheProperties.getInitialCapacity())
                .expireAfterWrite(Duration.ofNanos(l2CacheProperties.getExpireAfterWriteNanos()))
                //.refreshAfterWrite(Duration.ofNanos(l2CacheProperties.getRefreshAfterWriteNanos()))
                .build();
        return cache;
    }

    @Override
    public String getName() {
        return this.name;
    }

    @Override
    public Object getNativeCache() {
        return this;
    }

    @Override
    @Nullable
    protected Object lookup(Object key) {
        if (this.cache instanceof LoadingCache<Object, Object> loadingCache) {
            return loadingCache.get(key);
        }
        Object value = this.cache.getIfPresent(key);
        if (value != null) {
            return value;
        }
        value = getFromRedis(key);
        if (value != null) {
            this.cache.put(key, value);
        }
        return value;
    }


    @Override
    @Nullable
    public <T> T get(Object key, Callable<T> valueLoader) {
        Object value = this.cache.get(key, new LoadFunction(valueLoader));
        return (T) fromStoreValue(value);
    }

    @Override
    public void put(Object key, Object value) {
        //全量缓存 VS  热点局部缓存， 全量缓存 --- 数据变动时写入缓存（flinkCDC）,查询只走缓存； 热点局部缓存 --- 数据变动时延时双删 或 延时双刷,查询先走缓存再查底层数据，回写；
        //延时双删 或 延时双刷---（延时双刷:即时查询数据库数据并写入redis,广播redis消息，清除caffine缓存）
        //此方法中无法执行查询数据库数据（缺少valueLoader），因此 在缓存一致性需求较严格的场景 只能适用 延时双删: this.evict(key);
        Object storeValue = toStoreValue(value);
        setRedisValue(key, storeValue);
        cacheMessageMulticaster.multicast(new CacheMessage(name, key, CacheOperation.EVICT));
        this.cache.put(key, storeValue);
    }


    @Override
    public void evict(Object key) {
        EvictTask evictTask = new EvictTask(key);
        evictTask.run();
        CacheScheduler.schedule(evictTask, 8000L);
    }

    public void invalidate(Object key) {
        this.cache.invalidate(key);
    }

    @Override
    public void clear() {
        this.cache.invalidateAll();
    }


    private class LoadFunction implements Function<Object, Object> {

        private final Callable<?> valueLoader;

        public LoadFunction(Callable<?> valueLoader) {
            Assert.notNull(valueLoader, "Callable must not be null");
            this.valueLoader = valueLoader;
        }

        @Override
        public Object apply(Object key) {
            try {
                Object value = getFromRedis(key);
                return value != null ? value : getSynchronized(key, valueLoader);
            } catch (Exception ex) {
                throw new ValueRetrievalException(key, this.valueLoader, ex);
            }
        }
    }

    class EvictTask implements Runnable {
        private final Object key;

        public EvictTask(Object key) {
            this.key = key;
        }

        @Override
        public void run() {
            removeFromRedis(key);
            cacheMessageMulticaster.multicast(new CacheMessage(name, key, CacheOperation.EVICT));
            //redisTemplate.convertAndSend(l2CacheConfigProperties.getSyncTopic(), new CacheMessage(name, key, CacheOperation.EVICT));
            cache.invalidate(key);
        }
    }

    @Nullable
    @SuppressWarnings("unchecked")
    private Object getSynchronized(Object key, Callable<?> valueLoader) throws Exception {
        lock.lock();
        try {
            Object value = getFromRedis(key);
            if (value != null) {
                return value;
            }
            value = toStoreValue(valueLoader.call());

            setRedisValue(key, value);

            return value;
        } finally {
            lock.unlock();
        }
    }


    private Object getFromRedis(Object key) {
        String redisKey = getRedisKey(key);
        Object value = null;
        if (redisDataType.equals(RedisDataType.HASH)) {
            value = redisTemplate.opsForHash().get(redisKey, key);
        } else {
            value = redisTemplate.opsForValue().get(redisKey);
        }
        return value;
    }

    private void setRedisValue(Object key, Object value) {
        Duration expire = this.redisTimeToLive;
        if (NullValue.INSTANCE.equals(value)) {
            Duration defaultExpire = Duration.ofSeconds(300);
            if (Objects.isNull(expire) || expire.compareTo(defaultExpire) > 0) {
                expire = defaultExpire;
            }
        }
        String redisKey = getRedisKey(key);
        if (redisDataType.equals(RedisDataType.HASH)) {
            redisTemplate.opsForHash().put(redisKey, key, value);
        } else {
            if (expire != null) {
                redisTemplate.opsForValue().set(redisKey, value, expire);
            } else {
                redisTemplate.opsForValue().set(redisKey, value);
            }
        }
    }

    private void removeFromRedis(Object key) {
        String redisKey = getRedisKey(key);
        if (redisDataType.equals(RedisDataType.HASH)) {
            redisTemplate.opsForHash().delete(redisKey, key);
        } else {
            redisTemplate.delete(redisKey);
        }
    }


    protected String getRedisKey(Object key) {
        StringBuilder builder = new StringBuilder();
        if (redisUseKeyPrefix && redisKeyPrefix != null && !redisKeyPrefix.isEmpty()) {
            builder.append(redisKeyPrefix);
        }
        builder.append(getName());
        if (redisDataType.equals(RedisDataType.STRING)) {
            builder.append(":").append(key);
        }
        return builder.toString();
    }


}
